package org.raftkv.raft.tracker;


import org.raftkv.commons.Triple;
import org.raftkv.commons.Tuple;
import org.raftkv.raft.pb.ConfState;
import org.raftkv.raft.quorum.AckedIndexer;
import org.raftkv.raft.quorum.JointConfig;
import org.raftkv.raft.quorum.MajorityConfig;
import org.raftkv.raft.quorum.VoteResult;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.function.BiConsumer;

public class ProgressTracker {

    private Config config;

    private ProgressMap progress;

    private Map<Long, Boolean> votes;

    private int maxInflight;

    private ProgressTracker() {

    }

    /**
     * 初始化一个 ProgressTracker
     */
    public static ProgressTracker makeProgressTracker(int maxInflight) {
        ProgressTracker p = new ProgressTracker();
        p.maxInflight = maxInflight;
        p.config = new Config();
        p.config.setVoters(new JointConfig(new MajorityConfig(), null));
        p.votes = new HashMap<>();
        p.progress = new ProgressMap();

        return p;
    }

    /**
     * 该函数返回当前激活的配置的 ConfState
     */
    public ConfState confState() {
        ConfState confState = new ConfState();
        Optional.ofNullable(this.config.getVoters().CONFIGS[0]).ifPresent((voters -> {
            confState.setVoters(voters.asList());
        }));
        Optional.ofNullable(this.config.getVoters().CONFIGS[1]).ifPresent((votersOutgoing -> {
            confState.setVotersOutgoing(votersOutgoing.asList());
        }));
        confState.setLearners(new MajorityConfig(this.config.getLearners()).asList());
        confState.setLearnersNext(new MajorityConfig(this.config.getLearnersNext()).asList());
        confState.setAutoLeave(this.config.isAutoLeave());

        return confState;
    }

    /**
     * 当（且仅当）当前配置中只有一个投票者（Leader）时，返回 true
     */
    public boolean isSingleton() {
        return this.config.getVoters().CONFIGS[0].size() == 1 && this.config.getVoters().CONFIGS[1].size() == 1;
    }

    public long committed() {
        return this.config.getVoters().committedIndex(new MatchAckIndexer(progress));
    }

    public void visit(BiConsumer<Long, ProgressMap> f) {
        List<Long> ids = new ArrayList<>(this.progress.getProgressMap().keySet());
        ids.sort(Long::compare);

        for (Long id : ids) {
            f.accept(id, this.progress);
        }
    }

    public boolean quorumActive() {
        Map<Long, Boolean> votes = new HashMap<>();
        visit((id, prs) -> {
            Progress pr = prs.getProgressMap().get(id);
            if (pr.isLearner()) {
                return;
            }
            votes.put(id, pr.isRecentActive());
        });

        return config.getVoters().voteResult(votes) == VoteResult.VoteWon;
    }

    public List<Long> voterNodes() {
        Set<Long> m = config.getVoters().ids();

        List<Long> nodes = new ArrayList<>(m);
        nodes.sort(Long::compare);

        return nodes;
    }

    public List<Long> learnerNodes() {
        if (config.getLearners() == null || config.getLearners().size() == 0) {
            return null;
        }

        List<Long> nodes = new ArrayList<>(config.getLearners());
        nodes.sort(Long::compare);

        return nodes;
    }

    public void resetVotes() {
        votes = new HashMap<>();
    }

    public void recordVote(Long id, boolean v) {
        boolean ok = votes.containsKey(id);
        if (!ok) {
            votes.put(id, v);
        }
    }

    public Triple<Integer, Integer, VoteResult> tallyVotes() {
        int granted = 0;
        int rejected = 0;

        for (Map.Entry<Long, Progress> entry : progress.getProgressMap().entrySet()) {
            Long id = entry.getKey();
            Progress pr = entry.getValue();
            if (pr.isLearner()) {
                continue;
            }
            Boolean v = votes.get(id);
            if (v == null) {
                continue;
            }

            if (v) {
                granted++;
            } else {
                rejected++;
            }
        }

        VoteResult result = config.getVoters().voteResult(votes);
        return new Triple<>(granted, rejected, result);
    }

    private class MatchAckIndexer extends ProgressMap implements AckedIndexer {

        public MatchAckIndexer(ProgressMap p) {
            super(p);
        }

        @Override
        public Tuple<Long, Boolean> ackedIndex(long voterId) {
            Progress pr = this.getProgressMap().get(voterId);
            if (pr == null) {
                return new Tuple<>(0L, false);
            }
            return new Tuple<>(pr.getMatch(), true);
        }

    }

    public Config getConfig() {
        return config;
    }

    public void setConfig(Config config) {
        this.config = config;
    }

    public ProgressMap getProgress() {
        return progress;
    }

    public void setProgress(ProgressMap progress) {
        this.progress = progress;
    }

    public Map<Long, Boolean> getVotes() {
        return votes;
    }

    public void setVotes(Map<Long, Boolean> votes) {
        this.votes = votes;
    }

    public int getMaxInflight() {
        return maxInflight;
    }

    public void setMaxInflight(int maxInflight) {
        this.maxInflight = maxInflight;
    }

}
